package cn.fz.simple;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.HashSet;
import java.util.Set;

/**
 * 拉模式
 */
public class PullConsumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("PullConsumer");
        consumer.setNamesrvAddr("localhost:9876");
        Set<String> topics = new HashSet<>();
        topics.add("Simple");
        topics.add("TestTopic");
        consumer.setRegisterTopics(topics);
        consumer.start();
        while (true) {
            consumer.getRegisterTopics().forEach(n->{
                try {
                    consumer.fetchSubscribeMessageQueues(n).forEach(mq -> {
                        try {
                            long offset = consumer.getOffsetStore().readOffset(mq, ReadOffsetType.READ_FROM_MEMORY);
                            if (offset < 0){
                                offset = consumer.getOffsetStore().readOffset(mq, ReadOffsetType.READ_FROM_STORE);
                            }
                            if (offset < 0){
                                offset = consumer.maxOffset(mq);
                            }
                            if (offset < 0){
                                offset = 0;
                            }
                            PullResult pullResult = consumer.pull(mq, "*", offset, 32);
                            System.out.println("消息拉取成功-"+pullResult);
                            switch (pullResult.getPullStatus()){
                                case FOUND:
                                    pullResult.getMsgFoundList().forEach(k-> System.out.println("消息消费成功-"+k));
                                    consumer.updateConsumeOffset(mq,pullResult.getNextBeginOffset());
                            }
                        } catch (MQClientException e) {
                            throw new RuntimeException(e);
                        } catch (RemotingException e) {
                            throw new RuntimeException(e);
                        } catch (MQBrokerException e) {
                            throw new RuntimeException(e);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    });
                } catch (MQClientException e) {
                    throw new RuntimeException(e);
                }
            });
        }
    }
}
